/*
 * Copyright 2018 data Artisans GmbH Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License. You may obtain a copy of the
 * License at http://www.apache.org/licenses/LICENSE-2.0 Unless required by applicable law or agreed
 * to in writing, software distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the License for the specific
 * language governing permissions and limitations under the License.
 */

package com.dataartisans.flinktraining.examples.datastream_java.basics;

import java.text.SimpleDateFormat;
import java.util.Date;

import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.java.tuple.Tuple;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.api.java.utils.ParameterTool;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.log4j.LogManager;
import org.apache.log4j.Logger;

import com.dataartisans.flinktraining.exercises.datastream_java.datatypes.TaxiRide;
import com.dataartisans.flinktraining.exercises.datastream_java.sources.TaxiRideSource;
import com.dataartisans.flinktraining.exercises.datastream_java.utils.ExerciseBase;

/**
 * Example that counts the rides for each driver.
 *
 * Parameters: -input path-to-input-file
 *
 * Note that this is implicitly keeping state for each driver. This sort of simple, non-windowed
 * aggregation on an unbounded set of keys will use an unbounded amount of state. When this is an
 * issue, look at the SQL/Table API, or ProcessFunction, or state TTL, all of which provide
 * mechanisms for expiring state for stale keys.
 */
public class RideCount {
    private static Logger logger = LogManager.getLogger(RideCount.class);

    public static void main(String[] args) throws Exception {

        ParameterTool params = ParameterTool.fromArgs(args);
        final String input = params.get("input", ExerciseBase.pathToRideData);

        final int maxEventDelay = 60;       // events are out of order by max 60 seconds
        final int servingSpeedFactor = 600; // events of 10 minutes are served every second

        // set up streaming execution environment
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // start the data generator
        DataStream<TaxiRide> rides = env
                .addSource(new TaxiRideSource(input, maxEventDelay, servingSpeedFactor));

        // map each ride to a tuple of (driverId, 1)
        DataStream<Tuple2<Long, Long>> tuples = rides
                .map(new MapFunction<TaxiRide, Tuple2<Long, Long>>() {
                    @Override
                    public Tuple2<Long, Long> map(TaxiRide ride) throws Exception {
                        return new Tuple2<Long, Long>(ride.driverId, 1L);
                    }
            });

        // partition the stream by the driverId
        KeyedStream<Tuple2<Long, Long>, Tuple> keyedByDriverId = tuples.keyBy(0);

        // by peter
        logger.info(keyedByDriverId);

        // count the rides for each driver
        DataStream<Tuple2<Long, Long>> rideCounts = keyedByDriverId.sum(1);

        // we could, in fact, print out any or all of these streams
        // rideCounts.print();
        // 添加时间，并输出到文件。
        rideCounts.map(new MapFunction<Tuple2<Long, Long>, String>() {
            @Override
            public String map(Tuple2<Long, Long> value) throws Exception {
                return String.format("%s: %s",
                    new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(new Date()),
                    value.toString());
            }
        }).writeAsText("/data/bigdata/forflink/count.out");


        // run the cleansing pipeline
        env.execute("Ride Count");
    }

    public static class Smoother
        extends RichMapFunction<Tuple2<String, Double>, Tuple2<String, Double>> {
        private ValueState<Double> status;

        @Override
        public void open(Configuration parameters) throws Exception {
            ValueStateDescriptor<Double> descriptor = new ValueStateDescriptor<Double>(
                "status-name", Double.class);
            status = getRuntimeContext().getState(descriptor);
            super.open(parameters);
        }

        @Override
        public Tuple2<String, Double> map(Tuple2<String, Double> value) throws Exception {
            // TODO Auto-generated method stub
            return null;
        }

    }
}
